# -*- coding: utf-8 -*-
import pendulum
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.exceptions import AirflowSkipException
from airflow.models import BaseOperator


@DeprecationWarning
class LatestOnlyMixin(LoggingMixin):
    """
    已弃用
    建议使用 utils.operators.mixins.condition_run_mixin.LatestOnlyMixin
    2021-02-01 后可删除本类
    """

    def is_latest(self, context):
        if context['dag_run'] and context['dag_run'].external_trigger:
            self.log.info("Externally triggered DAG_Run: allowing execution to proceed.")
            return True

        now = pendulum.utcnow()
        left_window = context['dag'].following_schedule(context['execution_date'])
        right_window = context['dag'].following_schedule(left_window)
        self.log.info(
            'Checking latest only with left_window: %s right_window: %s now: %s',
            left_window, right_window, now
        )
        return left_window < now <= right_window

    def pre_execute(self, context):
        if self.is_latest(context) is False:
            # 非最新一次 dag_run，跳过运行
            self.log.info('Not latest execution, skipping.')
            raise AirflowSkipException

        # 获取最顶层的 Operator 类型，执行 pre_execute
        cls = next(c for c in self.__class__.__bases__ if issubclass(c, BaseOperator))
        cls.pre_execute(self, context=context)
